package com.lost.octopus.es.processor.scheduler;

import com.sunsharing.share.common.base.exception.ShareBusinessException;
import com.sunsharing.share.common.base.exception.ShareResponseCode;
import com.sunsharing.share.common.collection.ListUtil;
import com.lost.octopus.es.processor.config.es.alias.AliasElasticsearchRepository;
import com.lost.octopus.es.processor.dao.mapper.mybatis.BaseAsyncMapper;
import com.lost.octopus.es.processor.entity.IdEntity;
import com.lost.octopus.es.processor.scheduler.entity.FullSyncInfo;
import com.lost.octopus.es.processor.util.ReflectUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.elasticsearch.common.StopWatch;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * 索引同步操作者
 * 1.继承{@link AbstractSyncScheduler} 同样的调度方法不能同时跑
 * 2.需要一些默认的泛型，用于对象注入；需要知道ID是什么类型的
 * @author zhangbh
 * 2022-12-13 17:26
 **/
@Log4j2
public abstract class EsSyncIndexOperator<T extends IdEntity<ID>, ID, A extends BaseAsyncMapper<T>, E extends AliasElasticsearchRepository<T, ID>>
    extends AbstractSyncScheduler {

    /**
     * mybatis同步mapper
     */
    @Autowired(required = false)
    private A asyncMapper;
    /**
     * es存储dao
     */
    @Autowired(required = false)
    private E esDao;

    @Resource
    protected EsSyncManagement management;

    @Autowired
    private RedisIndexRebuildRecorder redisIndexRebuildRecorder;

    @Lazy
    @Resource
    private RedisJobRecorder redisJobRecorder;

    @Value("${es.main.tab.id.page.size:10000}")
    protected Integer esMainTabIdSize;

    /**
     * 获取同步查询的mapper
     * @return mapper
     */
    protected A getMapper() {
        return this.asyncMapper;
    }

    /**
     * 获取主键自增类型仓库
     * @return repository
     */
    public E getEsRepository() {
        return esDao;
    }

    /**
     * 重载索引 从{@link com.lost.octopus.es.processor.web.controller.base.BaseServiceController} 迁移
     * @param ids es上的数据id
     */
    public int reloadByIds(List<String> ids) {
        int count = 0;
        if (CollectionUtils.isEmpty(ids)) {
            return count;
        }
        StopWatch stopWatch = new StopWatch();
        // 如果提交的数据过大，直接分页存储，防止内存占用过大
        List<List<String>> lists = ListUtil.partition(ids, 1000);
        for (int i = 0; i < lists.size(); i++) {
            stopWatch.start(businessKey().getLabel() + " 重载索引第" + (i + 1) + "页");
            List<T> saveData = getByEsIds(lists.get(i));
            if (!CollectionUtils.isEmpty(saveData)) {
                // 判断索引是否在重建
                if (redisIndexRebuildRecorder.indexHasLocked(saveData.get(0).getClass())) {
                    throw new ShareBusinessException(ShareResponseCode.BIZ_ILLEGAL.getCode(), "索引正在重建，请稍后在保存数据");
                }
                getEsRepository().saveAll(saveData);
            }
            stopWatch.stop();
        }
        if (log.isDebugEnabled()) {
            log.debug("\n{} 从在索引数据size {}，过程耗时{} id {}",
                businessKey().getLabel(), ids.size(), stopWatch.prettyPrint(), ids);
        }
        return ids.size();
    }

    /**
     * 在保存前做的动做，可以对其进行扩展
     * @param list 数据
     */
    protected void beforeSave(List<T> list) {
    }

    public List<T> getByEsIds(List<String> esIds) {
        // todo 单元测试
        if (CollectionUtils.isEmpty(esIds)) {
            return new ArrayList<>();
        }
        Object id = esIds.get(0);
        try {
            boolean esIdTypeInteger = idType().equals(Integer.class);
            if (!esIdTypeInteger) {
                throw new RuntimeException(businessKey().getLabel() + "索引尝试reload，当前索引id为String，无法直接调用 findWillSyncInfo 需要重写本方法。或者拆解支持重载!!!");
            }
            Integer.valueOf(id.toString());
        } catch (Exception e) {
            throw new RuntimeException(businessKey().getLabel() + "索引尝试reload，当前索引id为String，无法直接调用 findWillSyncInfo 需要重写本方法。或者拆解支持重载!!!");
        }
        List<Integer> mainTblIds = esIds.stream().map(Integer::valueOf).collect(Collectors.toList());
        return findWillSyncInfo(mainTblIds);
    }

    /**
     * 查询实际要存储的实体
     * @param mainTabIds ID范围
     * @return 存储的实体
     */
    public List<T> findWillSyncInfo(List<Integer> mainTabIds) {
        List<T> data = getMapper().findWillSyncInfo(mainTabIds);
        if (CollectionUtils.isEmpty(data)) {
            return data;
        }
        // preSave doSomething
        beforeSave(data);
        return data;
    }

    /**
     * 查询需要存储到es的所有ID
     * @return 所有ID
     */
    public List<Integer> scrollMainTabIds(Integer maxId) {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start(this.businessKey().name() + "_" + maxId);
        // 实际查询代码
        List<Integer> ids = getMapper().scrollMainTabIds(maxId, esMainTabIdSize);
        stopWatch.stop();
        if (log.isDebugEnabled()) {
            log.debug("{} maxId {} pageSize {} 当前查询id耗时={}", this.businessKey().getLabel(), maxId, esMainTabIdSize, stopWatch.prettyPrint());
        }
        return ids;
    }

    /**
     * 获取实现类实例类型
     * @return clazz
     */
    Class<T> entityClass() {
        return (Class<T>) ReflectUtil.getSuperClassGenericType(this.getClass(), 0);
    }

    /**
     * 获取esId存储的类型
     */
    public Class<ID> idType() {
        return (Class<ID>) ReflectUtil.getEsIdType(ReflectUtil.getSuperClassGenericType(this.getClass(), 1));
    }

    /**
     * 传入当前处理的索引，满足以下条件则返回额外需要处理的索引
     * 1.存在全量调度
     * 2.和当前处理的不同
     */
    Optional<IndexCoordinates> diffAlias(IndexCoordinates timestampAlias) {
        // todo 单元测试
        Optional<FullSyncInfo> opt = redisJobRecorder.runningJob(businessKey());
        if (!opt.isPresent()) {
            return Optional.empty();
        }
        FullSyncInfo fullJobInfo = opt.get();
        String jobName = fullJobInfo.getJobName();
        boolean equals = Objects.equals(timestampAlias.getIndexName(), jobName);
        if (equals) {
            return Optional.empty();
        }
        log.info("{} 存在全量调度，获取索引 {}", businessKey().getLabel(), jobName);
        return Optional.of(getEsRepository().getTimestampAlias(jobName));
    }

    /**
     * 保存，若有全量索引，也保存
     */
    public void saveAll(List<T> records, IndexCoordinates timestampAlias) {
        getEsRepository().saveAll(records, timestampAlias);
        // 检查，若有全量，则也存储
        Optional<IndexCoordinates> opt = diffAlias(timestampAlias);
        opt.ifPresent(indexCoordinates -> getEsRepository().saveAll(records, indexCoordinates));
    }

    /**
     * 删除，若有全量索引，也删除
     */
    public void deleteAllById(List<ID> allIds, IndexCoordinates timestampAlias) {
        getEsRepository().deleteAllById(allIds, timestampAlias);
        // 检查，若有全量，则也处理
        Optional<IndexCoordinates> opt = diffAlias(timestampAlias);
        opt.ifPresent(indexCoordinates -> getEsRepository().deleteAllById(allIds, indexCoordinates));
    }
}
